package org.apache.streampark.console.core.service.impl;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.metadata.OrderItem;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.streampark.console.base.util.CommonUtils;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.enums.FlinkAppState;
import org.apache.streampark.console.core.mapper.ApplicationMapper;
import org.apache.streampark.console.core.service.StatisticsService;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class StatisticsServiceImpl extends ServiceImpl<ApplicationMapper, Application>
        implements StatisticsService {
    @Resource
    private ApplicationMapper applicationMapper;

    @Override
    public Map<String, Integer> getTaskStatistics(Long teamId) {
        //FlinkAppState
        //运行中
        Integer[] runningStatus = {FlinkAppState.RUNNING.getValue()};
        int runningCount = applicationMapper.getStatisticsTask(teamId, runningStatus);
        //ADDED、CANCELED 待运行
        Integer[] beRunStatus = {FlinkAppState.ADDED.getValue(),FlinkAppState.CANCELED.getValue()};
        int beRunCount = applicationMapper.getStatisticsTask(teamId, beRunStatus);
        //有异常 FAILING(6),FAILED(7)
        Integer[] failStatus = {FlinkAppState.FAILING.getValue(),FlinkAppState.FAILED.getValue()};
        int failCount = applicationMapper.getStatisticsTask(teamId, failStatus);
        //已完成 FINISHED(10)
        Integer[] finishedStatus = {FlinkAppState.FAILING.getValue(),FlinkAppState.FINISHED.getValue()};
        int finishedCount = applicationMapper.getStatisticsTask(teamId, finishedStatus);

        Map<String, Integer> map = new HashMap<>();
        map.put("runningCount", runningCount);
        map.put("beRunCount", beRunCount);
        map.put("failCount", failCount);
        map.put("finishedCount", finishedCount);
        return map;
    }

    @Override
    public List<Application> queryRunningTask(Long teamId) {
        Application appParam = new Application();
        appParam.setTeamId(teamId);
        //状态为运行中
        Integer[] newArray = {FlinkAppState.STARTING.getValue(),FlinkAppState.CANCELED.getValue(),FlinkAppState.RUNNING.getValue()};
        appParam.setStateArray(newArray);

        Page<Application> page = new Page<>();
        page.setCurrent(1);
        page.setSize(5);
        IPage<Application> resultPage = applicationMapper.page(page,appParam);
        List<Application> list = resultPage.getRecords();
        list = list.stream().sorted(Comparator.comparing(Application::getModifyTime).reversed()).collect(Collectors.toList());
        return list;
    }

    @Override
    public List<Application> queryCompletedTask(Long teamId) {
        Application appParam = new Application();
        appParam.setTeamId(teamId);
        //状态为已完成
        Integer[] newArray = {FlinkAppState.FINISHED.getValue()};
        appParam.setStateArray(newArray);

        Page<Application> page = new Page<>();
        page.setCurrent(1);
        page.setSize(5);
        IPage<Application> resultPage = applicationMapper.page(page,appParam);
        List<Application> list = resultPage.getRecords();
        list = list.stream().sorted(Comparator.comparing(Application::getModifyTime).reversed()).collect(Collectors.toList());
        return list;
    }
}
